package com.surfilter.massdata.spark.util.ipcheck;

import java.io.IOException;
import java.util.Arrays;
import java.util.HashMap;
import java.util.Iterator;
import java.util.List;
import java.util.Map;
import java.util.Map.Entry;

import com.google.common.collect.Maps;
import com.surfilter.massdata.spark.bean.ipcheck.ErrIp;
import com.surfilter.massdata.spark.bean.ipcheck.IpBean;
import com.surfilter.massdata.spark.bean.ipcheck.ObjectInfo;


/** 
* @author  hutao
* @date 创建时间：2017年5月24日 下午5:15:49 
* @Description：
* @version 1.0 
* 
**/
public class IpSourceConflictChecker {
	
	
	
	
	public static ObjectInfo sourceConflictCheck(List<IpBean> reduceSourceList) throws IOException, InterruptedException{
		ObjectInfo objectInfo=new ObjectInfo();
//		if(reduceSourceList.size()>1){
//			for(IpBean b:reduceSourceList){
//				System.err.println(b.getQsip()+"\t"+b.getZzip()+"\t"+b.getBbdw_id()+"\t"+b.getLyid()+"\t"+b.getId());
//			}
//		}
		
		Map<Long, IpBean> bbdwMap = new  HashMap<Long, IpBean>();
		Map<Long, IpBean> lydwMap = new  HashMap<Long, IpBean>();
		
		for(IpBean source : reduceSourceList){
			if(source.getIsPass() == null || source.getIsPass() == true){
				Long bbdw = source.getBbdw_id();
				Long lydw = source.getLyid();
				if(bbdwMap.get(bbdw) == null){
					if(lydwMap.get(lydw) == null){
						bbdwMap.put(bbdw, source);
						lydwMap.put(lydw, source);
						continue;
					}else{
						//冲突：a来源于c b来源于c
//						redisDao.hset(IpCheckConstants.CONFLICT_DATA_KEY, key.toString(), "");
//						log.info("chcker78冲突IP:" + newIp);
						IpBean conflictA = source;
						IpBean conflictB = lydwMap.get(lydw);
//						source.setIsPass(false);
//						conflictB.setIsPass(false);
						bbdwMap.remove(bbdw);    //bbdw不是空的吗，为啥要remove???
						lydwMap.remove(lydw);
						ErrIp errIp=new ErrIp();
						//这里做一个比较判断是因为这样：比如我当前conflictA是4142,比对到conflictB是4143,那么key就是4142_4143
						//但是有可能reduce下一轮遍历先存储的conflictA是4143,比对到的conflictB是4142,那么key就是4143_4142了 
						//所以这里规范一致的拼装属性
						if(conflictA.getBbdw_id() > conflictB.getBbdw_id()){
							errIp.setBbdws(conflictA.getBbdw_id() + "," + conflictB.getBbdw_id());
							errIp.setRegionIds(conflictA.getStorageType() + "_" + conflictA.getTflag() + "_" + conflictA.getId() + "," + conflictB.getStorageType() + "_" + conflictB.getTflag() + "_" + conflictB.getId());
							errIp.setJsonStr(conflictA.getConflictMsg()+"\001"+conflictB.getConflictMsg());
						}else{
							errIp.setBbdws(conflictB.getBbdw_id() + "," + conflictA.getBbdw_id());
							errIp.setRegionIds(conflictB.getStorageType() + "_" + conflictB.getTflag() + "_" + conflictB.getId() + "," + conflictA.getStorageType() + "_" + conflictA.getTflag() + "_" + conflictA.getId());
							errIp.setJsonStr(conflictB.getConflictMsg()+"\001"+conflictA.getConflictMsg());
						}
						if(conflictA.getStorageType().equals(IpCheckConstants.BEIAN)){
							errIp.setConflictType(ConflictType.IP_SOURCE_CONFLICT_ERROR.getTypeValue());
							errIp.setStorageType(IpCheckConstants.BEIAN);
						}else if(conflictA.getStorageType().equals(IpCheckConstants.UPLOAD)){
							errIp.setConflictType(ConflictType.IP_SOURCE_CONFLICT_ERROR2.getTypeValue());
							errIp.setStorageType(IpCheckConstants.UPLOAD);
						}
						
						objectInfo.setResult(false);
						objectInfo.setErrInfo(errIp);
						return objectInfo;
					}
				}else{
					 //冲突：a来源于b,c来源于d 同 a来源于b、a来源于d、a来源于e冲突
					 boolean isTheSame = bbdwMap.get(bbdw).getLyid().equals(source.getLyid());
//					 conflictA.setIsPass(false);
//					 conflictB.setIsPass(false);
					 if(!isTheSame){
//						 redisDao.hset(IpCheckConstants.CONFLICT_DATA_KEY, key.toString(), "");
						 IpBean conflictA = bbdwMap.get(bbdw);
						 IpBean conflictB = source;
						 bbdwMap.remove(bbdw);  //这个地方可以remove的了吗？？？
						 ErrIp errIp=new ErrIp();
						 errIp.setBbdws(conflictA.getBbdw_id().toString());
						 if(conflictA.getId() > conflictB.getId()){
							 errIp.setRegionIds(conflictA.getStorageType() + "_" + conflictA.getTflag() + "_" + conflictA.getId() + "," + conflictB.getStorageType() + "_" + conflictB.getTflag() + "_" + conflictB.getId());
							 errIp.setJsonStr(conflictA.getConflictMsg()+"\001"+conflictB.getConflictMsg());
						 }else{
							 errIp.setRegionIds(conflictB.getStorageType() + "_" + conflictB.getTflag() + "_" + conflictB.getId() + "," + conflictA.getStorageType() + "_" + conflictA.getTflag() + "_" + conflictA.getId());
							 errIp.setJsonStr(conflictB.getConflictMsg()+"\001"+conflictA.getConflictMsg());
						 }
							if(conflictA.getStorageType().equals(IpCheckConstants.BEIAN)){
								errIp.setConflictType(ConflictType.IP_SOURCE_CONFLICT_ERROR.getTypeValue());
								errIp.setStorageType(IpCheckConstants.BEIAN);
							}else if(conflictA.getStorageType().equals(IpCheckConstants.UPLOAD)){
								errIp.setConflictType(ConflictType.IP_SOURCE_CONFLICT_ERROR2.getTypeValue());
								errIp.setStorageType(IpCheckConstants.UPLOAD);
							}
							objectInfo.setResult(false);
							objectInfo.setErrInfo(errIp);
							return objectInfo;
					 }// end isTheSame
				}
			}else{
				continue;
			}
		}
		
		if(bbdwMap.size() !=1 && bbdwMap.size() > 0){
			Iterator<Entry<Long, IpBean>> iter1 = bbdwMap.entrySet().iterator();
			Iterator<Entry<Long, IpBean>> iter2 = null;
			while(iter1.hasNext()){
				Entry<Long, IpBean> entry = iter1.next();
				Long bbdw = entry.getKey();
				IpBean iterBean = entry.getValue();
				iter2 = lydwMap.entrySet().iterator();
				while(iter2.hasNext()){
					Entry<Long, IpBean> entry2 = iter2.next();
					Long lydw = entry2.getKey();
					IpBean iterBean2 = entry2.getValue();
						if(bbdw.equals(lydw)){     //这个地方做何解???
							iterBean.setIsPass(true);
							iterBean2.setIsPass(true);
							break;
						}
						//如果finalBbdwMap的size为1 是会把他的状态重置为true的
						iterBean.setIsPass(false);
				}
			}
			
			Map<Long, IpBean> finalBbdwMp = Maps.newHashMap();
			iter1 = bbdwMap.entrySet().iterator();
			while(iter1.hasNext()){
				Entry<Long, IpBean> entry = iter1.next();
				Long newip = entry.getKey();
				IpBean iterBean = ConvertUtil.copy(new IpBean(), entry.getValue());
				if(!new Boolean(true).equals(iterBean.getIsPass())){
					finalBbdwMp.put(newip, iterBean);
				}
			}
			
			if(finalBbdwMp.size() != 1 && finalBbdwMp.size() > 0){
//				redisDao.hset(IpCheckConstants.CONFLICT_DATA_KEY, key.toString(), "");
//				log.info("chcker173冲突IP:" + newIp);
				iter1 = finalBbdwMp.entrySet().iterator();
				String regionIds = "";
				String bbdws = "";
				String storageType = "";
				String tfFlag = "";
				String confinctMsgs="";
				long[] bbdwArr = new long[finalBbdwMp.size()];
				int index = 0;
				Map<Long, Map<String,String>> bbdwAndIdMap = Maps.newHashMap();
				while(iter1.hasNext()){
					IpBean ipBean = iter1.next().getValue();
//					ipBean.setIsPass(false);
					storageType = ipBean.getStorageType();
					tfFlag = ipBean.getTflag();
					bbdwArr[index] = ipBean.getBbdw_id();
					Map<String,String> localmap=Maps.newHashMap();
					localmap.put("id", ipBean.getId().toString());
					localmap.put("msg", ipBean.getConflictMsg());
					bbdwAndIdMap.put(bbdwArr[index], localmap);
					index++;
				}
				
				Arrays.sort(bbdwArr);
				for(long bbdw : bbdwArr){
					bbdws += bbdw + ",";
					regionIds += storageType + "_" + tfFlag + "_" + bbdwAndIdMap.get(bbdw).get("id") + ",";
					confinctMsgs+=bbdwAndIdMap.get(bbdw).get("msg") + "\001";
				}
				 ErrIp errIp=new ErrIp();
				 errIp.setBbdws(bbdws.substring(0, bbdws.length() - 1));
				 errIp.setRegionIds(regionIds.substring(0, regionIds.length() - 1));
				 errIp.setJsonStr(confinctMsgs.substring(0, confinctMsgs.length() - 1));
				if(storageType.equals(IpCheckConstants.BEIAN)){
					errIp.setConflictType(ConflictType.IP_SOURCE_CONFLICT_ERROR.getTypeValue());
					errIp.setStorageType(IpCheckConstants.BEIAN);
				}else{
					errIp.setConflictType(ConflictType.IP_SOURCE_CONFLICT_ERROR2.getTypeValue());
					errIp.setStorageType(IpCheckConstants.UPLOAD);
				}
				objectInfo.setResult(false);
				objectInfo.setErrInfo(errIp);
				return objectInfo;
			}else if(finalBbdwMp.size() == 1){
				//重置标志位
				Entry<Long, IpBean> entry = finalBbdwMp.entrySet().iterator().next();
				IpBean resetIpBean = entry.getValue();
				long resetIpBeanId = resetIpBean.getId();
				for(IpBean ipBean : reduceSourceList){
					if(ipBean.getId() == resetIpBeanId)
						ipBean.setIsPass(true);
				}
			}
		}
		objectInfo.setResult(true);
		return objectInfo;
	}
	

	


}
